
#include <stdio.h>
#include <stdarg.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <pthread.h>
#include "list.h"
#include "queue.h"

#define LOCK_LIST(l) pthread_mutex_lock(&((l)->m_mutex))
#define UNLOCK_LIST(l) pthread_mutex_unlock(&((l)->m_mutex))


struct mr_queue_s { 
	pthread_mutex_t m_mutex;
	pthread_cond_t th_cond;
	mr_list_p plist;
	u_char state;
};


int mr_queue_init(mr_queue_p *pqueue,mr_list_data_free freefunc){
	
	*(pqueue)= (mr_queue_p)malloc(sizeof(mr_queue_t));
	memset(*pqueue,0,sizeof(mr_queue_t));
	pthread_mutex_init(&((*pqueue)->m_mutex),NULL);
	pthread_cond_init(&((*pqueue)->th_cond),NULL);
	 mr_list_init(&((*pqueue)->plist),freefunc);
	return 0;
}
int mr_queue_push(mr_queue_p pqueue,void *pdata,int datasize){

	LOCK_LIST(pqueue);
	mr_list_push(pqueue->plist,pdata,datasize);
	pthread_cond_signal(&(pqueue->th_cond));
	UNLOCK_LIST(pqueue);
	return 0;
}
int mr_queue_pop(mr_queue_p pqueue,void **pdata,int * datasize , struct timespec *timeout){

	 struct timespec ts;
    if( pqueue->state )
        return 1;
    clock_gettime(CLOCK_REALTIME, &ts);
    ts.tv_sec += timeout->tv_sec;
    if( (ts.tv_nsec += timeout->tv_nsec) > 1000000000 ) {
        ts.tv_nsec -= 1000000000;
        ++ts.tv_sec;
    }
    LOCK_LIST(pqueue);
    while (mr_list_size(pqueue->plist) == 0) {
        if( pthread_cond_timedwait(&(pqueue->th_cond), &(pqueue->m_mutex), &ts) ) {
            UNLOCK_LIST(pqueue);
            return 1;
        }
        if(pqueue->state ) {
            UNLOCK_LIST(pqueue);
            return 1;
        }
    }
	
	 mr_list_pop(pqueue->plist,pdata,datasize);
	
            UNLOCK_LIST(pqueue);
	
	return 0;
}
int mr_queue_close(mr_queue_p pqueue){
	
	LOCK_LIST(pqueue);
//	mr_list_close(pqueue->plist);
	pqueue->state=1;
	pthread_cond_broadcast(&(pqueue->th_cond));
	UNLOCK_LIST(pqueue);
	return 0;
}
int mr_queue_destroy(mr_queue_p pqueue){

	pthread_cond_destroy(&(pqueue->th_cond));
	pthread_mutex_destroy(&(pqueue->m_mutex));
	mr_list_destroy(pqueue->plist);
	free(pqueue);
	return 0;
}


